﻿#include "pch.h"

#ifdef _WIN64
#ifdef _DEBUG
#pragma comment(lib, "boost_filesystem-vc143-mt-gd-x64-1_88.lib")
#pragma comment(lib, "boost_log-vc143-mt-gd-x64-1_88.lib")
#pragma comment(lib, "boost_log_setup-vc143-mt-gd-x64-1_88.lib")
#pragma comment(lib, "boost_thread-vc143-mt-gd-x64-1_88.lib")
#pragma comment(lib, "boost_atomic-vc143-mt-gd-x64-1_88.lib")
#pragma comment(lib, "boost_json-vc143-mt-gd-x64-1_88.lib")
#pragma comment(lib, "boost_locale-vc143-mt-gd-x64-1_88.lib")
#pragma comment(lib, "boost_stacktrace_windbg-vc143-mt-gd-x64-1_88.lib")
#else
#pragma comment(lib, "boost_filesystem-vc143-mt-x64-1_88.lib")
#pragma comment(lib, "boost_log-vc143-mt-x64-1_88.lib")
#pragma comment(lib, "boost_log_setup-vc143-mt-x64-1_88.lib")
#pragma comment(lib, "boost_thread-vc143-mt-x64-1_88.lib")
#pragma comment(lib, "boost_atomic-vc143-mt-x64-1_88.lib")
#pragma comment(lib, "boost_json-vc143-mt-x64-1_88.lib")
#pragma comment(lib, "boost_locale-vc143-mt-x64-1_88.lib")
#endif
#else
#ifdef _DEBUG
#pragma comment(lib, "boost_filesystem-vc143-mt-gd-x32-1_88.lib")
#pragma comment(lib, "boost_log-vc143-mt-gd-x32-1_88.lib")
#pragma comment(lib, "boost_log_setup-vc143-mt-gd-x32-1_88.lib")
#pragma comment(lib, "boost_thread-vc143-mt-gd-x32-1_88.lib")
#pragma comment(lib, "boost_atomic-vc143-mt-gd-x32-1_88.lib")
#pragma comment(lib, "boost_json-vc143-mt-gd-x32-1_88.lib")
#pragma comment(lib, "boost_locale-vc143-mt-gd-x32-1_88.lib")
#else
#pragma comment(lib, "boost_filesystem-vc143-mt-x32-1_88.lib")
#pragma comment(lib, "boost_log-vc143-mt-x32-1_88.lib")
#pragma comment(lib, "boost_log_setup-vc143-mt-x32-1_88.lib")
#pragma comment(lib, "boost_thread-vc143-mt-x32-1_88.lib")
#pragma comment(lib, "boost_atomic-vc143-mt-x32-1_88.lib")
#pragma comment(lib, "boost_json-vc143-mt-x32-1_88.lib")
#pragma comment(lib, "boost_locale-vc143-mt-x32-1_88.lib")
#endif
#endif

#pragma comment(lib, "redis++.lib")
#ifdef _DEBUG
#pragma comment(lib, "hiredisd.lib")
#else
#pragma comment(lib, "hiredis.lib")
#endif

std::mutex thread_mutex;

// https://www.boost.org/doc/libs/1_88_0/libs/log/doc/html/log/detailed/utilities.html#log.detailed.utilities.setup.settings
namespace fs = std::filesystem;
namespace logging = boost::log;
namespace sinks = boost::log::sinks;
namespace expr = boost::log::expressions;
namespace attrs = boost::log::attributes;
namespace keywords = boost::log::keywords;

// 自定义 Redis sink
class RedisSink : public sinks::basic_formatted_sink_backend<char, sinks::synchronized_feeding>
{
public:
    RedisSink(const std::string &redis_host, int32_t redis_port, const std::string &channel, const std::string &password = "")
        : channel(channel) // 100ms超时
    {
        redisManager_ = boost::make_shared<RedisManager>(redis_host, redis_port);
        try
        {
            auto reply = redisManager_->ping();
            std::cout << "Redis connection verified. PING reply: " << reply << std::endl;
            // 启动一个线程来处理消息发布
            worker_thread_ = std::thread(&RedisSink::worker, this);
#ifdef _WIN32
            SetThreadPriority(worker_thread_.native_handle(), THREAD_PRIORITY_HIGHEST);
#endif
        }
        catch (const sw::redis::TimeoutError &e)
        {
            std::cerr << "Redis TimeoutError error: " << e.what() << std::endl;
        }
        catch (const sw::redis::IoError &e)
        {
            std::cerr << "Redis IoError error: " << e.what() << std::endl;
        }
        catch (const std::exception &e)
        {
            std::cerr << "General error: " << e.what() << std::endl;
        }
    }

    ~RedisSink()
    {
        stop_ = true;
        cv_.notify_all();
        if (worker_thread_.joinable())
        {
            worker_thread_.join();
        }
    }

    void consume(const logging::record_view &rec, const string_type &formatted_message)
    {
        try
        {
            push_message(formatted_message); // 如果 push 失败会抛异常
        }
        catch (const std::exception &e)
        {
            std::cerr << "Failed to push message to queue: " << e.what() << std::endl;
        }
    }
    void push_message(const std::string &message)
    {
        std::unique_lock<std::mutex> lock(mutex_);
        queue_.push(message);
        cv_.notify_one();
    }

    std::string pop_message()
    {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait(lock, [this]
                 { return stop_ || !queue_.empty(); });
        if (stop_)
        {
            return "";
        }
        std::string message = queue_.front();
        queue_.pop();
        return message;
    }

private:
    void worker()
    {
        stop_ = false;
        while (!stop_)
        {
            std::string message = pop_message();
            if (!message.empty())
            {
                try
                {
                    redisManager_->publish(channel, UTF8(message));
                }
                catch (const std::exception &e)
                {
                    std::cerr << "Redis Pipeline Error: " << e.what() << std::endl;
                }
            }
            else
            {
                std::this_thread::sleep_for(100ms);
            }
        }
    }

    boost::shared_ptr<RedisManager> redisManager_;
    std::string channel;
    std::thread worker_thread_;
    std::queue<std::string> queue_;
    std::mutex mutex_;
    std::condition_variable cv_;
    bool stop_;
};

// 初始化日志配置
void init_logging(std::string appPath)
{
    if (appPath.empty())
    {
        appPath = getExecutableDirectory();
    }

    // 配置日志核心
    auto core = logging::core::get();

    // 配置文件输出
    fs::path logFilePath = fs::path(appPath) / "BridgeAnalysis_%5N.log";

    auto time_fmt = expr::format_date_time<boost::posix_time::ptime>("TimeStamp", "%Y-%m-%d %H:%M:%S.%f");

    // 创建通用格式
    auto fmt = expr::stream << std::setw(8) << std::setfill(' ') << std::right << expr::attr<uint32_t>("LogSequenceNumber") << " . " << time_fmt << " [" << logging::trivial::severity << "]"
                            << " [" << expr::attr<attrs::current_thread_id::value_type>("ThreadID") << "]: " << expr::smessage;

    auto file_sink = logging::add_file_log(keywords::file_name = logFilePath.string(), keywords::rotation_size = 100 * 1024 * 1024, keywords::auto_flush = true, keywords::open_mode = std::ios::app, keywords::format = fmt);

    file_sink->locked_backend()->set_file_collector(sinks::file::make_collector(keywords::target = appPath, keywords::max_size = 100 * 1024 * 1024, keywords::min_free_space = 10 * 1024 * 1024));
    file_sink->locked_backend()->scan_for_files();

    // 修改控制台日志的初始化部分
    auto console_sink = logging::add_console_log(std::cout, keywords::format = fmt);

    // 添加 Redis sink
    std::string redisHost = ConfigINI::GetInstance().GetString("RedisHost", "127.0.0.1");
    int32_t redisPort = ConfigINI::GetInstance().GetInt("RedisPort", 6379);
    STD_COUT << "redisHost:" << redisHost << STD_ENDL;
    STD_COUT << "redisPort:" << redisPort << STD_ENDL;

    auto redis_sink_backend = boost::make_shared<RedisSink>(redisHost, redisPort, "logging:channel:BridgeAnalysis");
    auto redis_sink = boost::make_shared<sinks::synchronous_sink<RedisSink>>(redis_sink_backend);
    redis_sink->set_formatter(fmt);
    core->add_sink(redis_sink);

    // 添加通用属性
    logging::add_common_attributes();
    core->add_global_attribute("ProcessID", attrs::current_process_id());
    core->add_global_attribute("ThreadID", attrs::current_thread_id());
    core->add_global_attribute("ProcessName", attrs::current_process_name());
    core->add_global_attribute("TimeStamp", attrs::local_clock());

    // 添加日志序号属性
    core->add_global_attribute("LogSequenceNumber", attrs::counter<uint32_t>(1));
}

// 获取当前毫秒时间戳
uint64_t now_ms()
{
    boost::posix_time::ptime epoch(boost::gregorian::date(1970, boost::gregorian::Jan, 1), boost::posix_time::time_duration(0, 0, 0));
    boost::posix_time::time_duration time_from_epoch = boost::posix_time::microsec_clock::universal_time() - epoch;
    return time_from_epoch.total_milliseconds();
}

// 获取当前进程ID
std::string getThreadId()
{
    std::string threadId = boost::lexical_cast<std::string>(boost::this_thread::get_id());
    return threadId;
}

// 格式化字符串
const std::string _S_(const char *format, ...)
{
    static char buf[10240];
    va_list ap;
    va_start(ap, format);
    vsnprintf(buf, sizeof(buf), format, ap);
    va_end(ap);
    return buf;
}

// UTF-8字符串转GBK字符串
std::string GBK(const std::string &strUTF8)
{
    return boost::locale::conv::from_utf(strUTF8.c_str(), std::string("gb2312"));
}

// GBK字符串转UTF-8字符串
std::string UTF8(const std::string &strGBK)
{
    return boost::locale::conv::to_utf<char>(strGBK.c_str(), std::string("gb2312"));
}

std::unordered_map<std::string, ThreadSafeReturnVars> __threadSafeReturnVarsMap;

// 清除错误代码和错误信息
void clear_error()
{
    std::lock_guard<std::mutex> lock(thread_mutex);
    std::string _threadId = getThreadId();
    auto _it = __threadSafeReturnVarsMap.find(_threadId);
    if (_it == __threadSafeReturnVarsMap.end())
    {
        __threadSafeReturnVarsMap.insert({_threadId, ThreadSafeReturnVars()});
        _it = __threadSafeReturnVarsMap.find(_threadId);
    }
    _it->second.vmessages.clear();
}

bool has_error()
{
    std::lock_guard<std::mutex> lock(thread_mutex);
    std::string _threadId = getThreadId();
    auto _it = __threadSafeReturnVarsMap.find(_threadId);
    if (_it == __threadSafeReturnVarsMap.end())
    {
        __threadSafeReturnVarsMap.insert({_threadId, ThreadSafeReturnVars()});
        _it = __threadSafeReturnVarsMap.find(_threadId);
    }
    if (_it->second.vcodes.size() > 0 && _it->second.vmessages.size() > 0)
    {
        return true;
    }
    return false;
}

// 设置动态库返回的错误代码和错误信息
void set_error(int32_t code, const char *message)
{
    std::lock_guard<std::mutex> lock(thread_mutex);
    std::string _threadId = getThreadId();
    auto _it = __threadSafeReturnVarsMap.find(_threadId);
    if (_it == __threadSafeReturnVarsMap.end())
    {
        __threadSafeReturnVarsMap.insert({_threadId, ThreadSafeReturnVars()});
        _it = __threadSafeReturnVarsMap.find(_threadId);
    }
    _it->second.vcodes.push_front(code);
    _it->second.vmessages.push_front(message);
    BOOST_LOG_TRIVIAL(error) << message;
}

// 获取动态库返回的错误代码和错误信息(最终一条)
void get_error(int32_t *pCode, char **pMessage)
{
    std::lock_guard<std::mutex> lock(thread_mutex);
    std::string _threadId = getThreadId();
    auto _it = __threadSafeReturnVarsMap.find(_threadId);
    if (_it == __threadSafeReturnVarsMap.end())
    {
        __threadSafeReturnVarsMap.insert({_threadId, ThreadSafeReturnVars()});
        _it = __threadSafeReturnVarsMap.find(_threadId);
    }
    if (_it->second.vcodes.size() > 0 && _it->second.vmessages.size() > 0)
    {
        if (pCode)
        {
            *pCode = _it->second.vcodes.front();
        }
        if (pMessage)
        {
            *pMessage = (char *)(_it->second.vmessages.front().c_str());
        }
    }
    else
    {
        if (pCode)
        {
            *pCode = 0;
        }
        if (pMessage)
        {
            *pMessage = nullptr;
        }
    }
}

// 获取动态库返回的多级错误代码和错误信息
void get_errors(int32_t *vcodes, char **vmessages, int32_t bufferCount, int32_t *realCount)
{
    std::lock_guard<std::mutex> lock(thread_mutex);
    std::string _threadId = getThreadId();
    auto _it = __threadSafeReturnVarsMap.find(_threadId);
    if (_it == __threadSafeReturnVarsMap.end())
    {
        __threadSafeReturnVarsMap.insert({_threadId, ThreadSafeReturnVars()});
        _it = __threadSafeReturnVarsMap.find(_threadId);
    }
    if (realCount)
    {
        *realCount = (int32_t)_it->second.vcodes.size();
    }
    auto it_code = _it->second.vcodes.begin();
    auto it_message = _it->second.vmessages.begin();
    int32_t i = 0;
    for (; it_code != _it->second.vcodes.end() && it_message != _it->second.vmessages.end(); ++it_code, ++it_message, ++i)
    {
        vcodes[i] = *it_code;
        vmessages[i] = (char *)it_message->c_str();
    }
}

// 获取运行程序的主目录
std::string getExecutableDirectory()
{
    std::string path;
#ifdef _WIN32
    char buffer[MAX_PATH];
    HMODULE hModule = GetModuleHandle(NULL);
    GetModuleFileNameA(hModule, buffer, MAX_PATH);
    path = buffer;
#else
    char buffer[MAX_PATH];
    ssize_t count = readlink("/proc/self/exe", buffer, MAX_PATH);
    if (count != -1)
    {
        path = std::string(buffer, count);
    }
#endif
    // 去除文件名，只保留目录
    size_t lastSlash = path.find_last_of("\\/");
    if (lastSlash != std::string::npos)
    {
        return path.substr(0, lastSlash);
    }

    return "";
}

// 获取动态库自身路径
std::string getModulePath()
{
    std::string path;
#ifdef _WIN32
    char buffer[MAX_PATH];
    HMODULE hModule = NULL;
    GetModuleHandleEx(GET_MODULE_HANDLE_EX_FLAG_FROM_ADDRESS, (LPCTSTR)getModulePath, &hModule);
    GetModuleFileNameA(hModule, buffer, MAX_PATH);
    path = buffer;
#else
    Dl_info dl_info;
    dladdr((void *)getModulePath, &dl_info);
    path = dl_info.dli_fname;
#endif
    size_t lastSlash = path.find_last_of("\\/");
    if (lastSlash != std::string::npos)
    {
        return path.substr(0, lastSlash);
    }
    return "";
}

// CRC64 表
const uint64_t CRC64_POLY = 0x42F0E1EBA9EA3693;
uint64_t crc64_table[256];

// 初始化 CRC64 表
void init_crc64_table()
{
    for (int32_t i = 0; i < 256; ++i)
    {
        uint64_t crc = i;
        for (int32_t j = 0; j < 8; ++j)
        {
            if (crc & 1)
            {
                crc = (crc >> 1) ^ CRC64_POLY;
            }
            else
            {
                crc >>= 1;
            }
        }
        crc64_table[i] = crc;
    }
}

// 计算 CRC64 值
uint64_t calculate_crc64(const char *data, size_t length)
{
    uint64_t crc = 0;
    for (size_t i = 0; i < length; ++i)
    {
        uint8_t index = static_cast<uint8_t>(crc ^ data[i]);
        crc = (crc >> 8) ^ crc64_table[index];
    }
    return crc;
}

ConfigINI &ConfigINI::GetInstance()
{
    static ConfigINI instance;
    return instance;
}

std::string ConfigINI::GetString(const std::string &key, std::string defaultValue)
{
    std::lock_guard<std::mutex> lock(config_mutex_);

    // 1. 从配置文件中查找
    auto it = config_map_.find(key);
    if (it != config_map_.end())
    {
        return it->second;
    }

    // 2. 安全获取环境变量（跨平台）
    char *env_val = nullptr;
    size_t len = 0;

#ifdef _WIN32
    // Windows安全版本
    _dupenv_s(&env_val, &len, key.c_str());
#else
    // Linux/macOS安全处理
    env_val = getenv(key.c_str());
    if (env_val)
        len = strlen(env_val) + 1;
#endif

    std::string result;
    if (env_val && len > 0)
    {
        result = std::string(env_val, len - 1); // 去掉终止符
    }

#ifdef _WIN32
    free(env_val); // 必须释放Windows分配的内存
#endif

    // 3. 返回默认值
    return result.empty() ? defaultValue : result;
}

int32_t ConfigINI::GetInt(const std::string &key, int32_t defaultValue)
{
    try
    {
        std::string val = GetString(key);
        if (!val.empty())
        {
            return std::stoi(val);
        }
    }
    catch (const std::exception &e)
    {
        std::cerr << "Invalid integer value for key " << key
                  << ": " << e.what() << std::endl;
    }
    return defaultValue;
}

double ConfigINI::GetDouble(const std::string &key, double defaultValue)
{
    try
    {
        std::string val = GetString(key);
        if (!val.empty())
        {
            return std::stod(val);
        }
    }
    catch (const std::exception &e)
    {
        std::cerr << "Invalid double value for key " << key
                  << ": " << e.what() << std::endl;
    }
    return defaultValue;
}

ConfigINI::ConfigINI(std::string iniFile)
{
    // 获取配置文件路径
    std::string exe_dir = getExecutableDirectory();
    if (iniFile.empty())
    {
        config_path_ = std::filesystem::path(exe_dir) / "BridgeAnalysis.ini";
    }
    else
    {
        config_path_ = iniFile;
    }
    // 加载配置文件
    LoadIniFile();
}

void ConfigINI::LoadIniFile()
{
    std::lock_guard<std::mutex> lock(config_mutex_);
    config_map_.clear();

    if (!std::filesystem::exists(config_path_))
        return;

    std::ifstream file(config_path_);
    std::string line;
    bool in_target_section = false;

    while (std::getline(file, line))
    {
        // 去除首尾空白
        line.erase(0, line.find_first_not_of(" \t"));
        line.erase(line.find_last_not_of(" \t") + 1);

        // 处理段落
        if (line.empty())
            continue;
        if (line[0] == '[')
        {
            in_target_section = (line.find("[BridgeAnalysis]") != std::string::npos);
            continue;
        }

        // 处理键值对
        if (in_target_section)
        {
            size_t pos = line.find('=');
            if (pos != std::string::npos)
            {
                std::string key = line.substr(0, pos);
                std::string value = line.substr(pos + 1);

                // 去除键值前后空白
                key.erase(key.find_last_not_of(" \t") + 1);
                value.erase(0, value.find_first_not_of(" \t"));
                value.erase(value.find_last_not_of(" \t") + 1);

                config_map_[key] = value;
            }
        }
    }
}

RedisManager::RedisManager(const std::string &host, int32_t port, int32_t socket_timeout_ms,
                           int32_t connect_timeout_ms)
    : host_(host), port_(port), should_stop_(false), socket_timeout_(socket_timeout_ms),
      connect_timeout_(connect_timeout_ms), last_active_time_(now_ms()), state_(ConnectionState::DISCONNECTED)
{
    connect();
}

RedisManager::~RedisManager()
{
    disconnect();
}

int64_t RedisManager::getLastActiveTime() const
{
    return last_active_time_.load();
}

RedisManager::ConnectionState RedisManager::getState()
{
    std::lock_guard<std::mutex> lock(mutex_);
    return state_;
}

void RedisManager::subscribe(const std::string &channel,
                             std::function<void(const std::string &, const std::string &)> callback)
{
    std::lock_guard<std::mutex> lock(mutex_);
    if (subscriber_)
    {
        try
        {
            subscriber_->subscribe(channel);
            subscriber_->on_message([this, callback](auto channel, auto msg)
                                    {
                    last_active_time_ = now_ms();  // 更新活跃时间
                    callback(channel, msg); });
            state_ = ConnectionState::CONNECTED; // 更新状态
        }
        catch (...)
        {
            state_ = ConnectionState::CONNECT_ERROR;
            throw;
        }
    }
}

bool RedisManager::consume()
{ // 修改：返回是否成功
    std::unique_lock<std::mutex> lock(mutex_);
    if (!subscriber_ || !redis_ || should_stop_)
        return false;

    try
    {
        lock.unlock();
        subscriber_->consume();
        last_active_time_ = now_ms(); // 更新活跃时间
        return true;
    }
    catch (const sw::redis::Error &e)
    {
        STD_CERR << "Redis consume error: " << e.what() << STD_ENDL;
        state_ = ConnectionState::CONNECT_ERROR;
        reconnect();
        return false;
    }
}

void RedisManager::disconnect()
{
    std::lock_guard<std::mutex> lock(mutex_);
    should_stop_ = true;
    subscriber_.reset();
    redis_.reset();
}

std::string RedisManager::ping()
{
    return redis_->ping();
}

int64_t RedisManager::publish(std::string theme, std::string message)
{
    std::lock_guard<std::mutex> lock(mutex_);
    int64_t ret = 0;
    for (int32_t i = 0; i < 3; i++)
    {
        try
        {
            ret = redis_->publish(theme, message);
            break;
        }
        catch (const sw::redis::Error &e)
        {
            ret = -1;
            STD_CERR << "Redis publish error: " << e.what() << STD_ENDL;
            state_ = ConnectionState::CONNECT_ERROR;
            reconnect();
        }
    }
    return ret;
}

void RedisManager::connect()
{
    std::lock_guard<std::mutex> lock(mutex_);
    try
    {
        sw::redis::ConnectionOptions opts;
        opts.host = host_;
        opts.port = port_;
        opts.connect_timeout = std::chrono::milliseconds(connect_timeout_);
        opts.socket_timeout = std::chrono::milliseconds(socket_timeout_); // 设置读超时
        opts.keep_alive = true;

        redis_ = std::make_shared<sw::redis::Redis>(opts);
        subscriber_ = std::make_shared<sw::redis::Subscriber>(redis_->subscriber());
        state_ = ConnectionState::CONNECTED;
        last_active_time_ = now_ms();
    }
    catch (...)
    {
        state_ = ConnectionState::CONNECT_ERROR;
        throw;
    }
}

void RedisManager::reconnect()
{
    for (int32_t i = 0; i < 3; ++i)
    { // 最多重试3次
        try
        {
            std::this_thread::sleep_for(std::chrono::seconds(1 << i)); // 指数退避
            connect();
            if (redis_)
                return;
        }
        catch (...)
        {
            // 忽略重连异常
        }
    }
    STD_CERR << "Failed to reconnect to Redis after multiple attempts" << STD_ENDL;
}